package com.luych.toolbox.data.service.job;

import com.luych.toolbox.common.enums.*;
import com.luych.toolbox.common.event.base.service.publisher.WebSocketEventPublisher;
import com.luych.toolbox.common.event.base.service.publisher.WebSocketLetterPublisher;
import com.luych.toolbox.common.event.base.service.publisher.WebSocketNoticePublisher;
import com.luych.toolbox.common.feign.intf.base.service.TaskFeign;
import com.luych.toolbox.common.feign.intf.gateway.center.FileFeign;
import com.luych.toolbox.common.feign.param.gateway.center.WebSocketEvent;
import com.luych.toolbox.common.service.tool.Cmd;
import com.luych.toolbox.data.service.constant.Constant;
import com.luych.toolbox.data.service.core.*;
import com.luych.toolbox.data.service.dao.*;
import com.luych.toolbox.data.service.entity.*;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.bus.BusProperties;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.core.io.Resource;
import org.springframework.scheduling.quartz.QuartzJobBean;
import org.springframework.transaction.support.TransactionTemplate;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.text.SimpleDateFormat;
import java.util.*;

public class DataProjectUpgradeJob extends QuartzJobBean {
    private static final Logger LOGGER = LoggerFactory.getLogger(DataProjectUpgradeJob.class);

    private final BusProperties busProperties;
    private final ApplicationEventPublisher applicationEventPublisher;

    private final TransactionTemplate transactionTemplate;

    private final DataUpgradeDao dataUpgradeDao;
    private final DataUpgradeDetailDao dataUpgradeDetailDao;
    private final DataProjectDao dataProjectDao;
    private final DataSourceDao dataSourceDao;
    private final DataAgentDao dataAgentDao;
    private final DataProjectRDataSourceDao dataProjectRDataSourceDao;

    private final DataBackupHistoryDao dataBackupHistoryDao;
    private final DataRecoveryHistoryDao dataRecoveryHistoryDao;

    private final FileFeign fileFeign;
    private final TaskFeign taskFeign;

    public DataProjectUpgradeJob(BusProperties busProperties,
                                 ApplicationEventPublisher applicationEventPublisher,
                                 TransactionTemplate transactionTemplate,
                                 DataUpgradeDao dataUpgradeDao,
                                 DataUpgradeDetailDao dataUpgradeDetailDao,
                                 DataProjectDao dataProjectDao,
                                 DataSourceDao dataSourceDao,
                                 DataAgentDao dataAgentDao,
                                 DataProjectRDataSourceDao dataProjectRDataSourceDao,
                                 DataBackupHistoryDao dataBackupHistoryDao,
                                 DataRecoveryHistoryDao dataRecoveryHistoryDao,
                                 FileFeign fileFeign,
                                 TaskFeign taskFeign) {
        this.busProperties = busProperties;
        this.applicationEventPublisher = applicationEventPublisher;
        this.transactionTemplate = transactionTemplate;
        this.dataUpgradeDao = dataUpgradeDao;
        this.dataUpgradeDetailDao = dataUpgradeDetailDao;
        this.dataProjectDao = dataProjectDao;
        this.dataSourceDao = dataSourceDao;
        this.dataAgentDao = dataAgentDao;
        this.dataProjectRDataSourceDao = dataProjectRDataSourceDao;
        this.dataBackupHistoryDao = dataBackupHistoryDao;
        this.dataRecoveryHistoryDao = dataRecoveryHistoryDao;
        this.fileFeign = fileFeign;
        this.taskFeign = taskFeign;
    }

    @Override
    protected void executeInternal(JobExecutionContext context) throws JobExecutionException {
        String id = context.getJobDetail().getJobDataMap().getString("id");
        // prepare
        WebSocketLetterPublisher letterPublisher = new WebSocketLetterPublisher(busProperties, applicationEventPublisher);
        WebSocketNoticePublisher noticePublisher = new WebSocketNoticePublisher(busProperties, applicationEventPublisher);
        WebSocketEventPublisher eventPublisher = new WebSocketEventPublisher(busProperties, applicationEventPublisher);

        String datetime = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date());

        // get data
        DataUpgrade dataUpgrade = dataUpgradeDao.findById(id).orElseThrow(() ->
                new JobExecutionException(Constant.DATA_UPGRADE_NOT_FOUND));
        DataProject dataProject = dataProjectDao.findById(dataUpgrade.getDataProjectId()).orElseThrow(() ->
                new JobExecutionException(Constant.DATA_PROJECT_NOT_FOUND));

        List<DataUpgradeDetail> dataUpgradeDetails = dataUpgradeDetailDao.findByDataUpgradeId(dataUpgrade.getId());
        Map<String, DataSource> dataUpgradeSources = new HashMap<>();
        Map<String, DataAgent> dataUpgradeAgents = new HashMap<>();
        for (DataUpgradeDetail dataUpgradeDetail : dataUpgradeDetails) {
            DataSource dataSource = dataSourceDao.findById(dataUpgradeDetail.getDataSourceId()).orElseThrow(() ->
                    new JobExecutionException(Constant.DATA_SOURCE_NOT_FOUND));
            dataUpgradeSources.put(dataUpgradeDetail.getId(), dataSource);

            Optional.of(dataUpgradeDetail).map(DataUpgradeDetail::getDataAgentId).flatMap(dataAgentDao::findById)
                    .ifPresent(dataAgent -> dataUpgradeAgents.put(dataUpgradeDetail.getId(), dataAgent));
        }

        // check data source status
        for (DataSource dataSource : dataUpgradeSources.values()) {
            if (!DataSourceStatus.NORMAL.equals(dataSource.getStatus())) {
                LOGGER.info("the data source [{}] status is {}, sleep 10 second and then retry...", dataSource.getName(), dataSource.getStatus());
                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    throw new JobExecutionException(e);
                }
                throw new JobExecutionException("need retry", true);
            }
        }

        // update status of data project and data upgrade
        transactionTemplate.executeWithoutResult(status -> {
            dataProject.setStatus(DataProjectStatus.UPGRADING);
            dataProjectDao.save(dataProject);

            dataUpgrade.setStartDate(new Date());
            dataUpgrade.setStatus(DataUpgradeStatus.RUNNING);
            dataUpgradeDao.save(dataUpgrade);
        });
        eventPublisher.send(new WebSocketEvent(WebSocketEvent.Type.DATA_PROJECT_UPGRADE_RUNNING, dataProject.getId()));

        String message = "数据项目" + dataUpgrade.getType().getValue() + "执行通知";

        // back datasource
        Map<String, DataBackupHistory> dataBackupHistories = new HashMap<>();
        String dataBackupError = null;
        String dataBackupTrace = null;
        for (DataUpgradeDetail dataUpgradeDetail : dataUpgradeDetails) {
            DataBackupHistory backupHistory = new DataBackupHistory();
            backupHistory.setId(UUID.randomUUID().toString());
            backupHistory.setCreateUserId(dataUpgrade.getCreateUserId());
            backupHistory.setCreateDate(new Date());
            if (DataUpgradeType.INIT.equals(dataUpgrade.getType())) {
                backupHistory.setName("数据项目[" + dataProject.getName() + "]初始化前自动备份");
                backupHistory.setType(DataBackupHistoryType.INIT);
            } else {
                backupHistory.setName("数据项目[" + dataProject.getName() + "]升级前自动备份，升级说明：" + dataUpgrade.getName());
                backupHistory.setType(DataBackupHistoryType.UPGRADE);
            }
            backupHistory.setStatus(DataBackupHistoryStatus.LOADING);
            backupHistory.setFile(UUID.randomUUID().toString());
            backupHistory.setDataSourceId(dataUpgradeDetail.getDataSourceId());
            backupHistory.setDataAgentId(dataUpgradeDetail.getDataAgentId());
            backupHistory.setOwnerId(dataUpgrade.getOwnerId());
            dataBackupHistoryDao.save(backupHistory);

            DataBackupHistoryExecuteJob backupJob = new DataBackupHistoryExecuteJob(busProperties,
                    applicationEventPublisher, transactionTemplate, dataBackupHistoryDao, dataSourceDao, dataAgentDao, taskFeign);
            backupJob.setId(backupHistory.getId());
            backupJob.executeInternal(null);

            backupHistory = dataBackupHistoryDao.findById(backupHistory.getId()).orElseThrow(() ->
                    new JobExecutionException(Constant.DATA_BACKUP_HISTORY_NOT_FOUND));
            if (DataBackupHistoryStatus.SUCCESS.equals(backupHistory.getStatus())) {
                dataBackupHistories.put(dataUpgradeDetail.getId(), backupHistory);
            } else {
                dataBackupError = dataUpgradeDetail.getErrorMessage();
                dataBackupTrace = dataUpgradeDetail.getErrorTrace();
                break;
            }
        }

        if (dataBackupError != null || dataBackupTrace != null) {
            String content = "数据项目[" + dataProject.getName() + "]于" + datetime + dataUpgrade.getType().getValue() + "异常，请及时处理。";
            noticePublisher.error(dataUpgrade.getOwnerId(), content);
            content += "\n\n\n错误信息：\n" + dataBackupError + "\n\n\n异常堆栈：\n" + dataBackupTrace;
            letterPublisher.error(dataUpgrade.getOwnerId(), message, content);
            eventPublisher.send(new WebSocketEvent(WebSocketEvent.Type.DATA_PROJECT_UPGRADE_ERROR, dataProject.getId()));
            return;
        }

        // exec upgrade
        Throwable dataUpgradeError = null;
        for (DataUpgradeDetail dataUpgradeDetail : dataUpgradeDetails) {

            if (dataUpgradeError != null) {
                dataUpgradeDetail.setStartDate(new Date());
                dataUpgradeDetail.setCompleteDate(new Date());
                dataUpgradeDetail.setStatus(DataUpgradeStatus.FAILED);
                dataUpgradeDetail.setErrorMessage("其他数据源升级失败");
                dataUpgradeDetail.setErrorTrace("其他数据源升级失败");
                dataUpgradeDetailDao.save(dataUpgradeDetail);
                continue;
            }

            DataSource dataSource = dataUpgradeSources.get(dataUpgradeDetail.getId());
            DataAgent dataAgent = dataUpgradeAgents.get(dataUpgradeDetail.getId());

            transactionTemplate.executeWithoutResult(status -> {
                dataSource.setStatus(DataSourceStatus.UPGRADING);
                dataSourceDao.save(dataSource);

                dataUpgradeDetail.setStartDate(new Date());
                dataUpgradeDetail.setStatus(DataUpgradeStatus.RUNNING);
                dataUpgradeDetailDao.save(dataUpgradeDetail);
            });
            eventPublisher.send(new WebSocketEvent(WebSocketEvent.Type.DATA_SOURCE_UPGRADE_RUNNING, dataSource.getId()));

            try {
                CoreConnection connection = CoreBuilder.buildConnection(dataSource);
                CoreDatabase database = connection.getDatabase();

                if (DataUpgradeType.INIT.equals(dataUpgrade.getType())) {
                    String sqlFile = dataAgent.getPath() + "/" + UUID.randomUUID().toString();

                    LOGGER.info("start upload file {} to {}@{}:{}", dataUpgradeDetail.getScript(),
                            dataAgent.getUser(), dataAgent.getHost(), sqlFile);
                    Resource sqlResource = fileFeign.get(dataUpgradeDetail.getScript());
                    Cmd.uploadRemote(dataAgent.getHost(), dataAgent.getPort(),
                            dataAgent.getUser(), dataAgent.getPass(), sqlResource.getInputStream(), sqlFile);

                    LOGGER.info("start clear data source {}", dataSource.getName());
                    for (CoreTable table : database.getTables()) {
                        connection.execute(table.getRemoveSql(CaseType.UNCHANGE));
                    }
                    for (CoreView view : database.getViews()) {
                        connection.execute(view.getRemoveSql(CaseType.UNCHANGE));
                    }

                    LOGGER.info("start init data source {}", dataSource.getName());
                    DataSourceType.Cmd cmd = dataSource.getType().getImportCmd(dataSource.getHost(), dataSource.getPort(),
                            dataSource.getUser(), dataSource.getPass(), dataSource.getDb(), sqlFile);
                    Cmd.executeRemote(dataAgent.getHost(), dataAgent.getPort(),
                            dataAgent.getUser(), dataAgent.getPass(), cmd.getCmd(),  cmd.getCodes());
                } else {
                    String sqlContent = dataUpgradeDetail.getScript();

                    LOGGER.info("start upgrade data source {}", dataSource.getName());
                    connection.execute(sqlContent);
                }

                transactionTemplate.executeWithoutResult(status -> {
                    dataSource.setStatus(DataSourceStatus.NORMAL);
                    dataSourceDao.save(dataSource);

                    dataUpgradeDetail.setCompleteDate(new Date());
                    dataUpgradeDetail.setStatus(DataUpgradeStatus.SUCCESS);
                    dataUpgradeDetailDao.save(dataUpgradeDetail);
                });
                eventPublisher.send(new WebSocketEvent(WebSocketEvent.Type.DATA_SOURCE_UPGRADE_SUCCESS, dataSource.getId()));
            } catch (Exception e) {
                LOGGER.error("error to upgrade data source [{}]: {}", dataSource.getName(), e.getMessage(), e);
                StringWriter writer = new StringWriter();
                e.printStackTrace(new PrintWriter(writer));

                transactionTemplate.executeWithoutResult(status -> {
                    dataSource.setStatus(DataSourceStatus.NORMAL);
                    dataSourceDao.save(dataSource);

                    dataUpgradeDetail.setCompleteDate(new Date());
                    dataUpgradeDetail.setStatus(DataUpgradeStatus.FAILED);
                    dataUpgradeDetail.setErrorMessage(e.getMessage());
                    dataUpgradeDetail.setErrorTrace(writer.toString());
                    dataUpgradeDetailDao.save(dataUpgradeDetail);
                });
                eventPublisher.send(new WebSocketEvent(WebSocketEvent.Type.DATA_SOURCE_UPGRADE_ERROR, dataSource.getId()));

                dataUpgradeError = e;
            }
        }

        if (dataUpgradeError != null) {
            StringWriter writer = new StringWriter();
            dataUpgradeError.printStackTrace(new PrintWriter(writer));

            String content = "数据项目[" + dataProject.getName() + "]于" + datetime + dataUpgrade.getType().getValue() + "异常，请及时处理。";
            noticePublisher.error(dataUpgrade.getOwnerId(), content);
            content += "\n\n\n错误信息：\n" + dataUpgradeError.getMessage() + "\n\n\n异常堆栈：\n" + writer.toString();
            letterPublisher.error(dataUpgrade.getOwnerId(), message, content);
            eventPublisher.send(new WebSocketEvent(WebSocketEvent.Type.DATA_PROJECT_UPGRADE_ERROR, dataProject.getId()));

            // rollback
            for (DataUpgradeDetail dataUpgradeDetail : dataUpgradeDetails) {
                DataSource dataSource = dataUpgradeSources.get(dataUpgradeDetail.getId());
                DataBackupHistory backupHistory = dataBackupHistories.get(dataUpgradeDetail.getId());

                DataRecoveryHistory recoveryHistory = new DataRecoveryHistory();
                recoveryHistory.setId(UUID.randomUUID().toString());
                recoveryHistory.setCreateUserId(dataUpgrade.getCreateUserId());
                recoveryHistory.setCreateDate(new Date());
                if (DataUpgradeType.INIT.equals(dataUpgrade.getType())) {
                    recoveryHistory.setName("数据项目[" + dataProject.getName() + "]初始化失败后自动恢复");
                    recoveryHistory.setType(DataRecoveryHistoryType.INIT);
                } else {
                    recoveryHistory.setName("数据项目[" + dataProject.getName() + "]升级失败后自动恢复，升级说明：" + dataUpgrade.getName());
                    recoveryHistory.setType(DataRecoveryHistoryType.UPGRADE);
                }
                recoveryHistory.setStatus(DataRecoveryHistoryStatus.LOADING);
                recoveryHistory.setDataBackupHistoryId(backupHistory.getId());
                recoveryHistory.setOwnerId(dataUpgrade.getOwnerId());
                dataRecoveryHistoryDao.save(recoveryHistory);

                DataRecoveryHistoryExecuteJob recoveryJob = new DataRecoveryHistoryExecuteJob(busProperties, applicationEventPublisher,
                        transactionTemplate, dataRecoveryHistoryDao, dataBackupHistoryDao, dataSourceDao, dataAgentDao, taskFeign);
                recoveryJob.setId(recoveryHistory.getId());
                recoveryJob.executeInternal(null);
            }

            transactionTemplate.executeWithoutResult(status -> {
                if (DataUpgradeType.INIT.equals(dataUpgrade.getType())) {
                    dataProject.setStatus(DataProjectStatus.UNINITIALIZED);
                } else {
                    dataProject.setStatus(DataProjectStatus.ENABLE);
                }
                dataProjectDao.save(dataProject);

                dataUpgrade.setCompleteDate(new Date());
                dataUpgrade.setStatus(DataUpgradeStatus.FAILED);
                dataUpgradeDao.save(dataUpgrade);
            });

            return;
        }

        transactionTemplate.executeWithoutResult(status -> {
            dataProject.setStatus(DataProjectStatus.ENABLE);
            dataProjectDao.save(dataProject);

            dataUpgrade.setCompleteDate(new Date());
            dataUpgrade.setStatus(DataUpgradeStatus.SUCCESS);
            dataUpgradeDao.save(dataUpgrade);
        });

        String content = "数据项目[" + dataProject.getName() + "]于" + datetime + dataUpgrade.getType().getValue() + "成功！";
        letterPublisher.success(dataUpgrade.getOwnerId(), message, content);
        noticePublisher.success(dataUpgrade.getOwnerId(), content);
        eventPublisher.send(new WebSocketEvent(WebSocketEvent.Type.DATA_PROJECT_UPGRADE_SUCCESS, dataProject.getId()));
    }
}
